package com.mashibing.rpc;
import com.mashibing.util.RabbitMQConnectionUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;

/**
 * 消费者类，用于从RabbitMQ队列中接收消息并处理
 */
public class Consumer {

    // 定义与生产者相关的队列名称
    public static final String QUEUE_PUBLISHER = "rpc_publisher";
    // 定义与消费者相关的队列名称
    public static final String QUEUE_CONSUMER = "rpc_consumer";

    /**
     * 消费消息的方法
     * 该方法会连接到RabbitMQ服务器，监听指定队列的消息，并在接收到消息后进行处理
     * @throws Exception 如果在连接、消费消息过程中发生错误，抛出此异常
     */
    @Test
    public void consume() throws Exception {
        //1. 获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();

        //2. 构建Channel
        Channel channel = connection.createChannel();

        //3. 构建队列
        channel.queueDeclare(QUEUE_PUBLISHER,false,false,false,null);
        channel.queueDeclare(QUEUE_CONSUMER,false,false,false,null);

        //4. 监听消息
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 处理接收到的消息
                System.out.println("消费者获取到消息：" + new String(body,"UTF-8"));
                // 准备响应信息
                String resp = "获取到了client发出的请求，这里是响应的信息";
                // 获取回复队列的名称
                String respQueueName = properties.getReplyTo();
                // 获取关联的ID，用于响应时关联请求
                String uuid = properties.getCorrelationId();
                // 构建响应的属性
                AMQP.BasicProperties props = new AMQP.BasicProperties()
                        .builder()
                        .correlationId(uuid)
                        .build();
                // 发送响应消息到回复队列
                channel.basicPublish("",respQueueName,props,resp.getBytes());
                // 确认消息已处理
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        // 开始消费消息
        channel.basicConsume(QUEUE_PUBLISHER,false,callback);
        System.out.println("开始监听队列");

        // 保持程序运行，持续监听
        System.in.read();
    }

}
